#!/usr/bin/python2.4
#
#Copyright 2009 Google Inc.
#
#Licensed under the Apache License, Version 2.0 (the "License");
#you may not use this file except in compliance with the License.
#You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
#Unless required by applicable law or agreed to in writing, software
#distributed under the License is distributed on an "AS IS" BASIS,
#WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#See the License for the specific language governing permissions and
#limitations under the License.
#

"""Tests db_utils module."""

import logging
import unittest
import zlib
from google.appengine.ext import db
import db_utils


class DbUtilsTest(unittest.TestCase):
  """Tests for the class DbHandler."""

  def setUp(self):
    """Set up method."""
    logging.info('TEST: ' + self.id())
    self.db_handler = db_utils.DbHandler()

  def testCreateNewVersion(self):
    """Tests the method CreateNewVersion."""
    version_id = self.db_handler.CreateNewVersion()
    self.assertEqual(version_id, 1)
    version_id_next = self.db_handler.CreateNewVersion()
    self.assertEqual(version_id + 1, version_id_next)
    version_id_3 = self.db_handler.CreateNewVersion()
    self.db_handler.RemoveVersion(version_id_next)
    self.assertEqual(version_id_3 + 1,
                     self.db_handler.CreateNewVersion())
    self.db_handler.RemoveVersion(1)
    self.db_handler.RemoveVersion(2)
    self.db_handler.RemoveVersion(3)
    self.db_handler.RemoveVersion(4)

  def testPublishVersion(self):
    """Tests the method PublishVersion."""
    version_id = self.db_handler.CreateNewVersion()
    version_id_next = self.db_handler.CreateNewVersion()
    page_map = {}
    test_html = '<html><body>Test For version 2</body></html>'
    page_map['test1.html'] = test_html
    self.db_handler.AddPages(page_map, version_id_next)
    version_id_3 = self.db_handler.CreateNewVersion()
    page_map = {}
    test_html = '<html><body>Test For version 3</body></html>'
    page_map['test1.html'] = test_html
    self.db_handler.AddPages(page_map, version_id_3)
    (status, published_id) = self.db_handler.PublishVersion(version_id_next)
    self.assert_(status)
    self.assertEqual(published_id, version_id_next)
    page_content = self.db_handler.GetPage('test1.html')
    self.assertEqual(page_content,
                     '<html><body>Test For version 2</body></html>')

    (status, published_id) = self.db_handler.PublishVersion(version_id_3)
    self.assert_(status)
    self.assertEqual(published_id, version_id_3)
    page_content = self.db_handler.GetPage('test1.html')
    self.assertEqual(page_content,
                     '<html><body>Test For version 3</body></html>')
    self.db_handler.RemoveVersion(version_id)
    self.db_handler.RemoveVersion(version_id_next)
    self.db_handler.RemoveVersion(version_id_3)

  def testGetAllVersions(self):
    """Tests the method GetAllVersions."""
    self.assertEqual(self.db_handler.GetAllVersions(), [])
    versions = []
    versions.append(self.db_handler.CreateNewVersion())
    delete_version = self.db_handler.CreateNewVersion()
    versions.append(self.db_handler.CreateNewVersion())
    versions.append(self.db_handler.CreateNewVersion())
    self.db_handler.RemoveVersion(delete_version)
    self.assertEqual(versions, self.db_handler.GetAllVersions())
    for version_id in versions:
      self.db_handler.RemoveVersion(version_id)
    self.assertEqual(self.db_handler.GetAllVersions(), [])

  def testAddImage(self):
    """Tests the method AddImage."""
    # Read Image file.
    f = open('test/data/logo_test.gif', 'r')
    data = f.read()
    f.close()
    status = self.db_handler.AddImage('/static/images/logo.gif',
                                      data)
    self.assert_(status)
    q = db.GqlQuery(
        'SELECT * FROM Image WHERE image_name=:1',
        '/static/images/logo.gif')
    self.assertEqual(q.count(), 1)
    r = q.fetch(1)
    self.assertEqual(r[0].image_name, '/static/images/logo.gif')
    self.assertEqual(r[0].image_data, data)
    db.delete(list(r))

  def testGetImage(self):
    """Tests the method GetImage."""
    f = open('test/data/logo_test.gif', 'r')
    data = f.read()
    f.close()
    status = self.db_handler.AddImage('/static/images/logo.gif',
                                      data)
    self.assert_(status)
    status = self.db_handler.AddImage('/static/images/logo1.gif',
                                      'data')
    self.assert_(status)
    self.assertEqual(
        data,
        self.db_handler.GetImage('/static/images/logo.gif'))
    self.assertEqual(
        'data',
        self.db_handler.GetImage('/static/images/logo1.gif'))
    self.assertEqual(
        None,
        self.db_handler.GetImage('/static/images/logo2.gif'))
    q = db.GqlQuery('SELECT * FROM Image')
    r = q.fetch(2)
    db.delete(list(r))

  def testAddPages(self):
    """Tests the method AddPages."""
    page_map = {}
    test_html = '<html><body>Test</body></html>'
    test_js = 'function(x) { document.createElement("div");'
    page_map['test1.html'] = test_html
    page_map['test1.js'] = test_js
    # Create a new version.
    version_id = self.db_handler.CreateNewVersion()
    self.assertEqual(version_id, 1)
    self.assert_(self.db_handler.AddPages(page_map, version_id))
    pages = db.GqlQuery(
        'SELECT * FROM Page WHERE version_id=:1', version_id)
    for page in pages:
      self.assert_(page.html_url in page_map.keys())
      self.assert_(
          zlib.decompress(str(page.html_content)) in page_map.values())
      self.assertEqual(version_id, page.version_id)
    # Cleanup.
    self.db_handler.RemoveVersion(1)

  def testGetPage(self):
    """Tests the method GetPage."""
    page_map = {}
    test_html = '<html><body>Test</body></html>'
    page_map['test1.html'] = test_html
    version_id = self.db_handler.CreateNewVersion()
    self.db_handler.AddPages(page_map, version_id)
    html_content = self.db_handler.GetPage('test1.html', version_id)
    self.assertNotEqual(html_content, None)
    self.assertEqual(html_content, test_html)
    invalid_content = self.db_handler.GetPage('test2.html', version_id)
    self.assertEqual(invalid_content, None)
    invalid_content = self.db_handler.GetPage('test1.html', 3)
    self.assertEqual(invalid_content, None)
    self.db_handler.RemoveVersion(version_id)

  def testRemovePages(self):
    """Tests the method RemovePages."""
    page_map = {}
    test_html = '<html><body>Test</body></html>'
    test_js = 'function(x) { document.createElement("div");'
    page_map['test1.html'] = test_html
    page_map['test1.js'] = test_js
    # Create a new version.
    version_id = self.db_handler.CreateNewVersion()
    self.db_handler.AddPages(page_map, version_id)
    self.db_handler.RemovePages(3)
    pages = db.GqlQuery('SELECT * FROM Page WHERE version_id=:1', version_id)
    self.assertEqual(pages.count(), 2)
    self.db_handler.RemovePages(version_id)
    pages = db.GqlQuery('SELECT * FROM Page WHERE version_id=:1', version_id)
    self.assertEqual(pages.count(), 0)
    self.db_handler.RemoveVersion(version_id)

  def testCreateParent(self):
    """Tests the method CreateParent."""
    par = self.db_handler.CreateParent(1)
    q = db.GqlQuery('SELECT * FROM PageParent WHERE version_id=1')
    r = q.fetch(1)
    self.assertEqual(par.version_id, r[0].version_id)
    self.db_handler.RemoveParent(1)

  def testGetParent(self):
    """Tests the method GetParent."""
    par = self.db_handler.GetParent(1)
    self.assertEqual(par.version_id, 1)
    self.db_handler.RemoveParent(1)

  def testRemoveParent(self):
    """Tests the method RemoveParent."""
    self.db_handler.CreateParent(1)
    self.db_handler.RemoveParent(1)
    q = db.GqlQuery('SELECT * FROM PageParent WHERE version_id=1')
    self.assertEqual(0, q.count())

  def testAddBulkMetadata(self):
    """Tests the method AddBulkMetadata."""
    version_id = self.db_handler.CreateNewVersion()
    parent = self.db_handler.CreateParent(version_id)
    m1 = {
        'row_id': 1,
        'doc_id': 'emfskdtifssa1',
        'version_id': version_id,
        'sc1': 'sc1',
        'sc2': 'sc2',
        'sc3': 'sc3',
        'title': 'title',
        'rest_metadata': '1     2       3       4',
        'pageparent': parent,
        }
    metadata = self.db_handler.CreateMetadata(m1['row_id'],
                                              m1['doc_id'],
                                              m1['version_id'],
                                              m1['sc1'],
                                              m1['sc2'],
                                              m1['sc3'],
                                              m1['title'],
                                              m1['rest_metadata'],
                                              m1['pageparent'])
    rows = []
    rows.append(metadata)
    self.db_handler.AddBulkMetadata(rows)
    metadata = self.db_handler.GetMetadataByRowNumber(1, version_id)
    self.assertEqual(metadata.row_id, m1['row_id'])
    self.assertEqual(metadata.id, m1['doc_id'])
    self.assertEqual(metadata.version_id, m1['version_id'])
    self.assertEqual(metadata.sorting_column1, m1['sc1'])
    self.assertEqual(metadata.sorting_column2, m1['sc2'])
    self.assertEqual(metadata.sorting_column3, m1['sc3'])
    self.assertEqual(metadata.title, m1['title'])
    self.assertEqual(metadata.rest_metadata, m1['rest_metadata'])
    self.db_handler.RemoveVersion(version_id)

  def testGetMetadataByRowNumber(self):
    """Tests the method GetMetadataByRowNumber."""
    parent = self.db_handler.CreateParent(1)
    version_id = self.db_handler.CreateNewVersion()
    m1 = {
        'row_id': 4,
        'doc_id': 'emfskdtifssa1',
        'version_id': version_id,
        'sc1': 'sc1',
        'sc2': 'sc2',
        'sc3': 'sc3',
        'title': 'title',
        'rest_metadata': '1     2       3       4',
        'pageparent': parent,
        }
    metadata = self.db_handler.CreateMetadata(m1['row_id'],
                                              m1['doc_id'],
                                              m1['version_id'],
                                              m1['sc1'],
                                              m1['sc2'],
                                              m1['sc3'],
                                              m1['title'],
                                              m1['rest_metadata'],
                                              m1['pageparent'])
    rows = []
    rows.append(metadata)
    self.db_handler.AddBulkMetadata(rows)
    metadata = self.db_handler.GetMetadataByRowNumber(1, version_id)
    self.assertEqual(metadata, None)
    metadata = self.db_handler.GetMetadataByRowNumber(4, version_id)
    self.assertEqual(metadata.row_id, m1['row_id'])
    self.assertEqual(metadata.id, m1['doc_id'])
    self.assertEqual(metadata.version_id, m1['version_id'])
    self.assertEqual(metadata.sorting_column1, m1['sc1'])
    self.assertEqual(metadata.sorting_column2, m1['sc2'])
    self.assertEqual(metadata.sorting_column3, m1['sc3'])
    self.assertEqual(metadata.title, m1['title'])
    self.assertEqual(metadata.rest_metadata, m1['rest_metadata'])
    self.db_handler.RemoveVersion(version_id)

  def testRemoveMetadata(self):
    """Tests the method RemoveMetadata."""
    version_id = self.db_handler.CreateNewVersion()
    parent = self.db_handler.CreateParent(version_id)
    rows = []
    rows.append(self.db_handler.CreateMetadata(1, '', version_id, 'sc1', 'sc2',
                                               'sc3', 'title', '1        2',
                                               parent))
    rows.append(self.db_handler.CreateMetadata(2, 'sds_eqdsS2', version_id,
                                               'sc1', 'sc2', 'sc3', 'title',
                                               '1 2', parent))
    rows.append(self.db_handler.CreateMetadata(3, '', version_id, 'sc1', 'sc2',
                                               'sc3', 'title', '1        2',
                                               parent))
    rows.append(self.db_handler.CreateMetadata(4, 'sds_eqdsS4', version_id,
                                               'sc1', 'sc2', 'sc3', 'title',
                                               '1 2', parent))
    self.db_handler.AddBulkMetadata(rows)
    self.db_handler.RemoveMetadata(version_id)
    self.assertEqual(self.db_handler.GetMetadataByRowNumber(1, version_id),
                     None)
    self.assertEqual(self.db_handler.GetMetadataByRowNumber(2, version_id),
                     None)
    self.assertEqual(self.db_handler.GetMetadataByRowNumber(3, version_id),
                     None)
    self.assertEqual(self.db_handler.GetMetadataByRowNumber(4, version_id),
                     None)
    self.db_handler.RemoveVersion(version_id)

  def testAddBulkGlobalData(self):
    """Tests the method AddBulkGlobalData."""
    version_id = self.db_handler.CreateNewVersion()
    parent = self.db_handler.CreateParent(version_id)
    rows = []
    rows.append(self.db_handler.CreateGlobalData(1, 'header image', 'logo.gif',
                                                 'header_img', version_id,
                                                 parent))
    rows.append(self.db_handler.CreateGlobalData(2, 'header width', '900',
                                                 'header_width', version_id,
                                                 parent))
    rows.append(self.db_handler.CreateGlobalData(3, 'header height', '100',
                                                 'header_height', version_id,
                                                 parent))
    rows.append(self.db_handler.CreateGlobalData(4, 'Footer Test', 'Copyright',
                                                 'footer_text', version_id,
                                                 parent))
    self.db_handler.AddBulkGlobalData(rows)
    globaldata = self.db_handler.GetGlobalDataByVariable('header_img',
                                                         version_id)
    self.assertEqual(globaldata.value, 'logo.gif')
    self.assertEqual(globaldata.type, 'header image')
    self.db_handler.RemoveVersion(1)

  def testRemoveGlobalData(self):
    """Tests the method RemoveGlobalData."""
    version_id = self.db_handler.CreateNewVersion()
    parent = self.db_handler.CreateParent(version_id)
    rows = []
    rows.append(self.db_handler.CreateGlobalData(1, 'header image', 'logo.gif',
                                                 'header_img', version_id,
                                                 parent))
    rows.append(self.db_handler.CreateGlobalData(2, 'header width', '900',
                                                 'header_width', version_id,
                                                 parent))
    rows.append(self.db_handler.CreateGlobalData(3, 'header height', '100',
                                                 'header_height', version_id,
                                                 parent))
    rows.append(self.db_handler.CreateGlobalData(4, 'Footer Test', 'Copyright',
                                                 'footer_text', version_id,
                                                 parent))
    self.db_handler.AddBulkGlobalData(rows)
    globaldata = self.db_handler.GetGlobalDataByVariable('header_img',
                                                         version_id)
    self.assertEqual(globaldata.value, 'logo.gif')
    self.db_handler.RemoveGlobalData(version_id)
    q = db.GqlQuery('SELECT * FROM GlobalData WHERE version_id=:1', version_id)
    self.assertEqual(q.count(), 0)
    self.db_handler.RemoveVersion(version_id)


def main(unused_argv):
  unittest.main()
